/***************************************************************************************
  *Copyright(C),2010-2016,Sumscope
  *FileName   : TableCache.cpp
  *Author     : scofined.qi
  *Version    : 1.0
  *Date       : 2016/06/02
  *Desc       : //用于说明此程序文件完成的主要功能
  *Relation   :
  *Other      : //其他内容说明
  *Function   : //主要函数列表，每条记录应包含函数名及功能简要说明
  *History    : //修改历史纪录列表，每条修改记录应包含修改日期，修改者以及修改内容简介
  ***************************************************************************************/
#include "TableCache.h"
#include "TableFile.h"
#include "dict.h"
#include "hash.h"
#include "KeyRecordIdHashMap.h"
#include <core/allocator/allocator_pool_manager.h>
#include <core/container/StringIntHashMap.h>
#include <core/strings/BaseString.h>
#include <core/strings/string_util.h>
#include <core/strings/string_convert.h>
#include <core/instance/singleton.h>
#include <core/time/time.h>
#include <core/Utility.h>
#include <core/FileSys.h>
#include <QTemporaryFile>
#include <QStandardPaths>
#include <QFile>
#include <QDir>

#ifdef WIN32
#else
#include <unistd.h>
#endif

/*
	[内存优化][scofined.qi]
	改进:
		NewKeyIdMap 中使用StringIntHashMap来存储字符串到int的映射关系

	优势:
	1.内存碎片少:原先各个string内存长短不一,内存有大有小,现在统一改造成
		256个字节,使得分配更为友好,碎片更少
	2.内存占用少:值int限定为2^24,符合当前使用场景,一个节点256字节放置多个键值对
		而且,支持将整个键值表与磁盘映射文件的快速交换,清空内存
	3.查找速度快:缓存友好,硬加速并行查找
*/
using namespace qb::base;
namespace qb{
	namespace ht{
		namespace{
			// BKDR Hash Function
			unsigned int BKDRHash(const char *str)
			{
				unsigned int seed = 131; // 31 131 1313 13131 131313 etc..
				unsigned int hash = 0;

				while (*str)
				{
					hash = (hash << 7) + (hash << 1) + hash + (*str++);
				}

				return (hash & 0x7FFFFFFF);
			}

			typedef unsigned int uint32_t;

			class MurmurHash3 {

#define 	ROLI(I,N) (int)((unsigned int)I>>N)
#define  ROTL32(dword, n) ((unsigned long) << (n) ^ ((unsigned long) >> (32 - (n))))

			public:
				/** Returns the MurmurHash3_x86_32 hash. */
                static int murmurhash3_x86_32(const unsigned char* data, int offset, int len, int seed) {

					const int c1 = 0xcc9e2d51;
					const int c2 = 0x1b873593;

					int h1 = seed;
					int roundedEnd = offset + (len & 0xfffffffc);  // round down to 4 byte block

					for (int i = offset; i < roundedEnd; i += 4) {
						// little endian load order
						int k1 = (data[i] & 0xff) | ((data[i + 1] & 0xff) << 8) | ((data[i + 2] & 0xff) << 16) | (data[i + 3] << 24);
						k1 *= c1;
						k1 = (k1 << 15) | (ROLI(k1, 17));  // ROTL32(k1,15);
						k1 *= c2;

						h1 ^= k1;
						h1 = (h1 << 13) | (ROLI(k1, 19));  // ROTL32(h1,13);
						h1 = h1 * 5 + 0xe6546b64;
					}

					// tail
					int k1 = 0;

					switch (len & 0x03) {
					case 3:
						k1 = (data[roundedEnd + 2] & 0xff) << 16;
						// fallthrough
					case 2:
						k1 |= (data[roundedEnd + 1] & 0xff) << 8;
						// fallthrough
					case 1:
						k1 |= (data[roundedEnd] & 0xff);
						k1 *= c1;
						k1 = (k1 << 15) | (ROLI(k1, 17));  // ROTL32(k1,15);
						k1 *= c2;
						h1 ^= k1;
					}

					// finalization
					h1 ^= len;

					// fmix(h1);
					h1 ^= ROLI(k1, 16);
					h1 *= 0x85ebca6b;
					h1 ^= ROLI(k1, 13);
					h1 *= 0xc2b2ae35;
					h1 ^= ROLI(k1, 16);

					return h1;
				}

			};
		}

        typedef qb::base::StringChunkMap StringChunkMap;
        typedef qb::base::StringView StringView;
		class OldHashMap : public IKeyIdMap{
			KeyRecordIdHashMap	m_map;
		public:
			OldHashMap(kDataType type, int bytes) :m_map(type,bytes)
			{}
			virtual ~OldHashMap(){}
			virtual void remove(const char* szKey){
				m_map.remove(szKey);
			}
			virtual void insert(const char* szKey, int len, int rid){
				m_map.insert(szKey, len, rid);
			}
			virtual int find(const char* szKey){
				return m_map.find(szKey);
			}
			virtual int GetMemoryBytes()const{
				return m_map.GetMemoryBytes();
			}
			virtual int GetCount()const{
				return m_map.GetCount();
			}
			virtual void Expand(int capcity, double keylen){}
			virtual bool Flush(){ return false; }
		};
		class NewKeyIdMap : public IKeyIdMap{
			typedef StringChunkMap ChunkMap;
			//typedef StdHashMap ChunkMap;
			kDataType		m_type;				// 数据类型
			int				m_init_keybytes;	// 键值表的初始平均尺寸
			int				m_calc_keybytes;	// 计算好的键的平均尺寸
			ChunkMap	*	m_keymap;			// 内存中的键值表
            QFile*			m_file;				// 刷新到磁盘上的文件句柄
			int				m_fsize;			// 文件尺寸
			BaseString		m_fname;			// 临时文件目录
		public:
			NewKeyIdMap(kDataType type, int bytes):
				m_type(type), m_init_keybytes(bytes), m_calc_keybytes(bytes),
				m_file(NULL), m_fsize(0), m_keymap(nullptr)
			{
				// bytes是key的最大长度,固定结构体中定义的key的长度
			}
			~NewKeyIdMap()
			{
                if (m_file!=NULL)
                {
                    FileSys::instance()->closeFile(m_file);
                    m_file = nullptr;
                }
                FileSys::instance()->removeFile(m_fname.c_str());
				if (m_keymap)
					delete m_keymap;
			}
			void remove(const char* szKey)
			{
				ChunkMap* pmap = m_keymap ? m_keymap : Reload();
				if (pmap)
					pmap->Erase(szKey);
			}
			virtual void GetAllRecord(vector<unsigned int>& records)override
			{
				ChunkMap* pmap = m_keymap ? m_keymap : Reload();
				if (pmap)
					pmap->GetValues(records);
			}
            inline void insert(const char* szKey,int len, int rid)
			{
				ChunkMap* pmap = m_keymap ? m_keymap:Reload();
				if (pmap)
					pmap->Put(StringView(szKey, len), rid);
			}
			int find(const char* szKey)
			{
				uint id = -1;
				ChunkMap* pmap = m_keymap ? m_keymap : Reload();
				if (pmap && pmap->Get(StringView(szKey), id))
					return (int)id;
				return -1;
			}

			int GetMemoryBytes()const{
				return m_keymap ? m_keymap->GetBytes() : 0;
			}
			int GetCount()const{
				return m_keymap ? m_keymap->GetCount() : 0;
			}
			void Expand(int capcity,double keylen)
			{
				// capcity是已经获取到的容量
				// 1 是0结尾1字节
				// 2 是tag+pos
				// 4 是StringView中的表示值和长度
				m_calc_keybytes = keylen + 1 + 2 + 4;
				double bytes = capcity*m_calc_keybytes;
				bool usePool = bytes < 64 * 1024 ? false : true;
				int oldSlot = m_keymap ? m_keymap->GetSlotCount() : 0;
				if (m_keymap == NULL)
					m_keymap = new ChunkMap();
				m_keymap->Expand(capcity, keylen, usePool);
				int newSlot = m_keymap->GetSlotCount();
			}
			bool Flush()
			{
				// 将内存中的哈希表内容刷新到磁盘文件中
                if (1)
				{
					return false;
				}
				if (m_keymap == nullptr)
					return true;
				if (m_file == nullptr)
				{
                    MakeTempPathFileName(m_fname);
                    bool bRet = FileSys::instance()->createFile(m_fname.c_str(), QIODevice::ReadWrite, m_file);
                    if (m_file == NULL)
					{
						return false;
					}
				}

				m_fsize = m_keymap->CalcFileSize();
                m_file->seek(0);
				char* buf = new char[m_fsize];
				int bytes = m_keymap->Save(buf,m_fsize);
                unsigned long dwWritten = 0;
				int membytes = m_keymap->GetBytes();
                bool suc = bytes == m_fsize ? m_file->write(buf, m_fsize) : false;

                m_file->seek(m_file->size());
				delete[] buf;
				if (suc)
					delete m_keymap;
				m_keymap = nullptr;
				return suc;
			}
		protected:
			void MakeTempPathFileName(BaseString& tempfile)
			{
                QString strFileName = QDir::tempPath() + QDir::separator() + QString("Cc");
                QTemporaryFile tmpFile(strFileName);
                tmpFile.open();
                tmpFile.close();
                tempfile = strFileName.toStdString().c_str();
			}
			ChunkMap* Reload()
			{
				if (m_keymap)
					return m_keymap;
                if (m_file != NULL && m_fsize>0)
				{
					char* buf = new char[m_fsize];
                    if (m_file->read(buf, m_fsize))
					{
						ChunkMap* pmap = new ChunkMap();
						if (pmap->Load(buf, m_fsize))
							m_keymap = pmap;
					}

					delete[] buf;
				}

				return m_keymap;
			}
		};
		//class 	KeyRecordIdMap
		//{
		//	typedef std::map<RecordKey, RecordId> StringIntMap;
		//	struct StringInt{
		//	public:
		//		static unsigned int hashFunction(const void *key)
		//		{
		//			const char* keystr = (const char*)key;
		//			//return BKDRHash(keystr);
		//			//return MurmurHash3::murmurhash3_x86_32((byte*)keystr, 0, strlen(keystr), 13);
		//			return leveldb::BloomHash(keystr, strlen(keystr));
		//		}
		//		static void * keyDup(void *privdata, const void *key)
		//		{
		//			const char* keystr = (const char*)key;
		//			int len = strlen(keystr);
		//			char* newkey = new char[len + 1];
		//			memcpy(newkey, keystr, len);
		//			newkey[len] = 0;
		//			return newkey;
		//		}
		//		static int keyCompare(void *privdata, const void *key1, const void *key2)
		//		{
		//			const char* keystr1 = (const char*)key1;
		//			const char* keystr2 = (const char*)key2;
		//			return strcmp(keystr1, keystr2) == 0 ? 1 : 0;
		//		}
		//		static void keyDestructor(void *privdata, void *key)
		//		{
		//			char* keystr = (char*)key;
		//			delete[] keystr;
		//		}
		//	};
		//	dictType			m_type;		// 
		//	dict*			m_dict;		// levedb实现的哈希表
		//	StringIntMap		m_idmap;		// std::map实现
		//	bool				m_usemap;	// 默认使用level的哈希表m_dict
		//	// 当此为true,则使用StringIntMap
		//public:
		//	KeyRecordIdMap() :m_usemap(true)
		//	{
		//		if (!m_usemap)
		//		{
		//			m_type.hashFunction = StringInt::hashFunction;
		//			m_type.keyDup = StringInt::keyDup;
		//			m_type.keyCompare = StringInt::keyCompare;
		//			m_type.keyDestructor = StringInt::keyDestructor;
		//			m_dict = dictCreate(&m_type, 0);
		//			//dictExpand(m_dict, 40480);
		//		}
		//	}
		//	~KeyRecordIdMap(){ if (!m_usemap)dictRelease(m_dict); }
		//	int find(const char* key)
		//	{
		//		if (m_usemap){
		//			StringIntMap::iterator it = m_idmap.find(key);
		//			return it != m_idmap.end() ? it->second : -1;
		//		}
		//		dictEntry *he = dictFind(m_dict, key);
		//		return he ? (int)dictGetVal(he) : -1;
		//	}
		//	bool insert(char* key, int rid)
		//	{
		//		if (m_usemap)	{
		//			m_idmap.insert(StringIntMap::value_type(key, rid)); return true;
		//		}
		//		return DICT_OK == dictAdd(m_dict, key, (void*)rid);
		//	}
		//};
		namespace{
			char* LoadFileContent(const char* fileName, int& fileSize)
			{
                fileSize = -1;
                QFile* file;
                FileSys::instance()->createFile(fileName, QIODevice::ReadWrite, file);
                if (file == nullptr)
                {
                    return NULL;
                }

                fileSize = FileSys::instance()->getFileSize(file);
                char* buf = new char[fileSize + 1];
                memset(buf, 0, fileSize + 1);
                FileSys::instance()->readFile(file, buf, fileSize);
                FileSys::instance()->closeFile(file);

                return buf;
			}
		}

        class BatchWriteCacheAllocator : public qb::base::Singleton<BatchWriteCacheAllocator>{
            qb::base::HeapAllocator		m_heap;
		public:
			BatchWriteCacheAllocator() :m_heap("BatchWriteCacheSingleton"){}
			void Free(void* mem, int bytes)
			{
				m_heap.Free(mem, bytes);
			}
			void* Alloc(int bytes)
			{
				return m_heap.Alloc(bytes);
			}
		};
		namespace{
			std::string MakeHeapName(TableFile* file){
                const char* fname = qb::base::StringUtil::GetShortFileName(file->GetFileName().c_str(), 1);
				return std::string("BatchWriteCache:") + (fname?fname:"");
			}
		}
		class BatchWriteCache{
			typedef map<RecordId, void*> RecordMap;//保证RecordId严格递增排序
            typedef qb::base::HeapAllocator Allocator;
			int			m_record_length;	// 单条记录的长度
			int			m_cache_max;		// 缓存超过这多就会触发刷新
			TableFile*	m_file;				// 映射文件
			RecordMap	m_records;			// 缓存的记录
			Allocator	m_heap;				// 分配缓存的堆
		public:
			BatchWriteCache(TableFile* file, int length) :
				m_heap(MakeHeapName(file).c_str()),
				m_cache_max(1024),
				m_file(file), m_record_length(length){
				if (m_record_length>0)
					m_cache_max = file->GetViewSize()*2 / m_record_length;
			}
			~BatchWriteCache()
			{
				Flush(true);
			}
			bool ReadCache(RecordId rid, char* record, int size)
			{
                BASSERT(record && size == m_record_length);
				RecordMap::const_iterator it = m_records.find(rid);
				if (it != m_records.end())
				{
					memcpy(record, it->second, size);
					return true;
				}
				return false;
			}
			bool WriteCache(RecordId rid, const void* record, int length)
			{
                BASSERT(m_record_length == length);
				if ((int)m_records.size() > m_cache_max)
					Flush(false);
				RecordMap::const_iterator it = m_records.find(rid);
				if (it != m_records.end())
				{
					memcpy(it->second, record, length);
				}
				else
				{
					void* cache = m_heap.Alloc(m_record_length);
					memcpy(cache, record, length);
					m_records[rid] = cache;
				}
				return true;
			}
			void Flush(bool force=true)
			{
				RecordMap::const_iterator it = m_records.begin();
				for (; it != m_records.end(); ++it)
				{
					void* record = it->second;
					m_file->Write(it->first, (const char*)record, m_record_length);
					m_heap.Free(record, m_record_length);
				}
				m_records.clear();
				if (force)
				{
					m_heap.Reset();
				}
			}
		};

		//void TableCache::test()
		//{
		//	int id = 0;
		//	std::map<string, int> m_keys;
		//	{	// 读取bondkeys 到数组中
		//		std::string fname = "D:\\test\\bondkeys.txt";
		//		int fsize = 0;
		//		char* fbuf = LoadFileContent(fname.c_str(), fsize);
		//		char* str = fbuf;

		//		while (str)
		//		{
		//			char* pend = strchr(str, '\r');
		//			if (pend)
		//			{
		//				*pend = 0; pend++;
		//				if (*pend == '\n')
		//				{
		//					*pend = 0; pend++;
		//				}
		//			};
		//			m_keys.insert(std::map<string, int>::value_type(string(str), id++));
		//			str = pend;
		//		}
		//		delete[] fbuf;
		//	}
		//	KeyRecordIdMap keys;

		//	for (std::map<string, int>::iterator it = m_keys.begin(); it != m_keys.end(); ++it)
		//	{
		//		if (!keys.insert((char*)it->first.c_str(), it->second))
		//		{
		//			BVIEW(0, 0, 0, "\ninsert key %s failed,id:%d", it->first.c_str(), it->second);
		//			keys.insert((char*)it->first.c_str(), it->second);
		//		}
		//	}

		//	for (std::map<string, int>::iterator it = m_keys.begin(); it != m_keys.end(); ++it)
		//	{
		//		int kid = keys.find(it->first.c_str());
		//		if (kid != it->second)
		//		{
		//			BVIEW(0, 0, 0, "\nkey:%s id:%d id2:%d", it->first.c_str(), it->second, kid);
		//		}
		//	}

		//}
		namespace{
			std::string GenerateGuidString()
			{
                return StringUtil::GenerateGuidString().toStdString();
			}
		}


		TableCache::TableCache(int recordSize,int headSize, kDataType type, int bytes) :
			m_record_length(recordSize), m_head_size(headSize),
			m_record_count(0),
			m_file(0),
			m_data_type(type),
			m_viewsize(4), m_increase_size(4),
			m_key_bytes(bytes),
			m_temp_file(false),
			m_use_quotamap(false),
			m_swap_record(NULL),
			m_batch_cache(NULL),
			m_tag("")
		{
			// 初始化创建
			if (m_use_quotamap)
				m_quotamap.Build(20480, 1024, recordSize);

			m_write_count = 0, m_swap_count = 0;
			m_key_recordmap = CreateKeyMap(type, bytes);
		}
		IKeyIdMap* TableCache::CreateKeyMap(kDataType type, int bytes)
		{
			//if (ss::base::cfg::GetConfigBool(kBaseConfigAssertBox))
			//	return new OldHashMap(type, bytes);
			return new NewKeyIdMap(type, bytes);
		}
		TableCache::~TableCache()
		{
			if (m_key_recordmap)	delete m_key_recordmap;
			if (m_file)				delete m_file;
			if (m_swap_record)		delete[] m_swap_record;
			if (m_batch_cache)		delete m_batch_cache; m_batch_cache = nullptr;
		}

		RecordId TableCache::GetRecordId(const char* key)
		{
			return m_key_recordmap->find(key);
		}
		bool TableCache::Read(RecordId rid, char* record, int size)
		{
            BASSERT(size == m_record_length && record);
            BASSERT(m_file);
			int id_max = GetFile()->GetRecordCount();
			if (rid >= id_max)
				return false;
			if (!m_delete_map.empty())
			{
				if (m_delete_map.find(rid) != m_delete_map.end())
					return  false;
			}
			if (m_use_quotamap && m_quotamap.Get(rid, record, false))
				return true;
			bool suc = false;
			if (m_batch_cache)
			{
				suc = m_batch_cache->ReadCache(rid, record, m_record_length);
			}
			if (!suc)
				suc = m_file->Read(rid, record, m_record_length);
			if (!suc)
				return false;
			// 将记录回写到限额缓存表m_quotamap中
			if (m_use_quotamap)
				Write(rid, record, size, "");
			return true;
		}
        int TableCache::GetMaxRecordId()const{
            BASSERT(m_file);
			return m_file->GetRecordCount()-1;
		}
		RecordId TableCache::Write(RecordId rid, const void* record, int size, const char* key, bool direct)
		{
			// rid = -1 表示直接插入到末尾，不修改
            BASSERT(record && size == m_record_length);
            int id_max = GetFile()->GetRecordCount();
            BASSERT((rid >= 0 && rid <= id_max) || (rid == -1&&key!=nullptr));
            BASSERT(m_file);
			bool insert = false;
			if (rid == -1)
			{
                BASSERT(key != nullptr);
				insert = true;
				m_record_count++;
				rid = id_max;
				m_key_recordmap->insert((char*)key,strlen(key), rid);
			}
			if (direct)
			{
				if (m_batch_cache)
				{
					m_batch_cache->WriteCache(rid, (const char*)record, m_record_length);
					m_file->SetRecordCount(rid);//不写入实际内容,稍后在写入,纯粹是更新个数
				}
				else
					m_file->Write(rid, (const char*)record, m_record_length);
				return rid;
			}

			// m_field_changed 用于指示当前哪个列发生了改变
			if (m_use_quotamap)
			{
				int swap_rid = -1;
				if (m_swap_record == 0)
				{
					m_swap_record = new char[m_record_length + 1];
					memset(m_swap_record, 0, m_record_length + 1);
				}
				memset(m_swap_record, 0, m_record_length);
				m_quotamap.Set(rid, record, &swap_rid, m_swap_record, insert);
				if (swap_rid >= 0)
				{
					m_swap_count++;
					m_file->Write(swap_rid, m_swap_record, m_record_length);
				}
			}


			if (m_batch_cache)
			{
				m_batch_cache->WriteCache(rid, (const char*)record, m_record_length);
				m_file->SetRecordCount(rid);//不写入实际内容,稍后在写入,纯粹是更新个数
			}
			else
				m_file->Write(rid, (const char*)record, m_record_length);
			//return rid;
			//m_file->Write(rid, (const char*)record, m_record_length);

			m_write_count++;
			return rid;
		}

		TableFile* TableCache::RecreateFile()
		{
			if (m_file)
			{
				delete m_file;
				m_file = NULL;
			}
			
			//m_quotamap.Build();
			m_record_count = 0;
			m_write_count = m_swap_count = 0;
			if (m_key_recordmap)	delete m_key_recordmap;
			m_key_recordmap = CreateKeyMap(m_data_type, m_key_bytes);
			m_delete_map.clear();
			if (m_batch_cache)delete m_batch_cache;
			m_batch_cache = nullptr;
			return CreateTableFile(true);
		}
		TableFile* TableCache::GetFile()
		{
			if (m_file == 0)
			{
				CreateTableFile(false);
			}
			return m_file;
		}
		TableFile* TableCache::CreateTableFile(bool reset)
		{
            BASSERT(m_file == NULL);
			bool temp = m_temp_file;
			if (m_name.empty())
				SetName("");
			m_file = new TableFile(m_record_length, m_head_size, m_tag.size() > 0 ? m_tag.c_str() : 0, m_viewsize, m_increase_size);
			m_file->SetDeleteFile(temp ? true : false);
			m_file->Initialize(m_name.c_str(),reset);
			if (m_batch_cache)delete m_batch_cache;
			if (0)
			{
				m_batch_cache = new BatchWriteCache(m_file, m_record_length);
			}
			return m_file;
		}
		void TableCache::SetQuotaListener(QuotaMap::Listener* listener)
		{
			m_quotamap.SetListener(listener);
		}

		void TableCache::SetCapcity(int capcity)
		{
			m_key_recordmap->Expand(capcity,m_keylen_stat.GetLength());
		}
		void TableCache::Flush()
		{
			m_key_recordmap->Flush();
			if (m_batch_cache)
			{
				m_batch_cache->Flush(true);
				delete m_batch_cache;
				m_batch_cache = nullptr;
			}
		}
		bool TableCache::ReadHead(char*buf, int size)
		{
			return GetFile()->ReadHead(buf, size);
		}
		bool TableCache::WriteHead(const char*buf, int size,int count)
		{
			return GetFile()->WriteHead(buf, size, count);
		}
		void TableCache::SetName(const std::string& fullFileName)
		{
			if (fullFileName.empty() || fullFileName.size() <= 0)
			{
                FileSys::instance()->makePathExist("cache");
				//string fname = GenerateGuidString() + ".fmap";
				//fname = string("./cache/") + fname;

#ifdef WIN32
                QString strProcess = QString("%1").arg(GetCurrentProcessId());
#else
                QString strProcess = QString("%1").arg(getpid());
#endif
                QString strName = QString("./cache/%1-%2.fmap").arg(strProcess, StringUtil::GenerateGuidString());
                m_name = strName.toStdString();
				return;
			}
			//BASSERT(fullFileName)
			// 检查文件名称合法,是否是全路径
			// 检查文件名是否可写
			// 检查对应文件是否存在

			m_name = fullFileName;
		}
        void TableCache::GetAllRecord(std::vector<unsigned int>& records)
		{
			m_key_recordmap->GetAllRecord(records);
		}
		void TableCache::BuildKeyMap(int count, int offset,int flagOffset)
		{
			/*
			m_file中总共有count条记录,其中关键列在每条记录中的偏移量为offset,大小为size
			关键列的类型必定是string类型,

			此接口只用于从映射文件中重建索引

			flagOffset:为1个字节的bool字段,默认值为0,当为1的时候,表示此条记录被删除
			*/
            BASSERT(offset >= 0 && offset + m_key_bytes <= m_record_length);
			
			bool ignore = !(flagOffset >= 0 && flagOffset < m_record_length);

			// 探测键值的长度,只进行一次
			HighResTime timer;
			for (int i = 0; i < count && !m_keylen_stat.IsFull(); i++)
			{
				const char* record = m_file->ReadDirect(i);
				if (record)
					m_keylen_stat.Add(strlen(record + offset));
			}
			
			double timeDetect = timer.elapsed(), timeExpand = 0;
			
            const char* fname = qb::base::StringUtil::GetShortFileName(m_name.c_str(), 1);
			timer = HighResTime();
			m_key_recordmap->Expand(count,m_keylen_stat.GetLength());
			timeExpand = timer.elapsed();

			if (m_batch_cache)
				m_batch_cache->Flush();
			m_delete_map.clear();

			//vector<BaseString> keys;
			//keys.reserve(count);
			timer = HighResTime();
			double readTime=0, insertTime = 0;
			class BatchInsert{
				struct KeyPair{ 
					uint		len : 8;		// 最大长度为256
					uint		id : 24;
					const char* str;
				};
				char*			m_buffer;
				int				m_capcity;
				int				m_used;
                std::vector<KeyPair>	m_keys;
				NewKeyIdMap*	m_map;
			public:
				BatchInsert(NewKeyIdMap* pmap) :m_map(pmap), m_used(0),
					m_capcity(16 * 1024)
				{
					m_buffer = new char[m_capcity];
					m_keys.reserve(1024);
				}
				~BatchInsert(){
					delete[] m_buffer;
				}
				void Add(const char* szkey,int id)
				{
					int len = strlen(szkey);
					if (len + 1 + m_used >= m_capcity)
						Flush();
					KeyPair kp; kp.id = id; kp.len = strlen(szkey);
					kp.str = m_buffer + m_used;
					memcpy(m_buffer + m_used, szkey, kp.len);
					m_buffer[m_used + kp.len] = '\0';
					m_used += kp.len + 1;
					m_keys.emplace_back(kp);
				}
				void Flush()
				{
					int nsize = m_keys.size();
					for (int i = 0; i < nsize; i++)
					{
						const KeyPair& kp = m_keys[i];
						m_map->insert(kp.str,kp.len,kp.id);
					}
					m_keys.clear();
					m_used = 0;
				}
			};

			//BatchInsert batch(m_key_recordmap);
			for (int i = 0; i < count; i++)
			{
				HighResTime tempTimer;
				const char* record = m_file->ReadDirect(i);
				readTime += tempTimer.elapsed();
				if (record == NULL)
				{
					m_delete_map.insert(i);
					continue;
				}
				tempTimer = HighResTime();
				if (ignore || !record[flagOffset])
				{
					//keys.emplace_back(BaseString(record + offset));
					int len = strlen(record + offset);
					const char* key = record + offset;
					m_key_recordmap->insert(key, len, i);
					//batch.Add(record + offset, i);
				}
				else
					m_delete_map.insert(i);
				insertTime += tempTimer.elapsed();
			}
			//batch.Flush();
			m_record_count = m_key_recordmap->GetCount();
			/*const char* pos = strrchr(m_name.c_str(), '\\');
			if (pos)
			{
			pos++;
			BaseString fname = os::fmtstr("./data/dict_%s.txt", pos);
			StringUtil::WriteLines(keys, fname);
			}*/
			static double totalRead = 0, totalInsert = 0;
			totalRead += timer.elapsed();
		}
		bool TableCache::RemoveKey(const char* key)
		{
			RecordId rid = GetRecordId(key);
			if (rid >= 0)
			{
				m_delete_map.insert(rid);
				m_key_recordmap->remove(key);
				return true;
			}
			return false;
		}

		int TableCache::GetMemoryBytes()const
		{
			int bytes = sizeof(*this);
			if (m_key_recordmap)
				bytes += m_key_recordmap->GetMemoryBytes();
			if (m_swap_record)
				bytes += m_record_length;
			if (m_use_quotamap)
				bytes += m_quotamap.GetMemoryUsed();
			//if (m_file)
			//	bytes += m_file->GetMemory();
			return bytes;
		}
	}//namespace ht
}//namespace qb
